#pragma once

#include <iostream>
#include <functional>
#include <memory>
#include <mutex>
#include <condition_variable>
#include <unordered_map>
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_logger.hpp"
#include "mq_consumer.hpp"


namespace bitmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;    
    //客户端只能收到两种响应类型
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string& cid, const muduo::net::TcpConnectionPtr& conn, 
            const ProtobufCodecPtr& codec)
            : _cid(cid)
            , _conn(conn)
            , _codec(codec)
        {}

        ~Channel()
        {
            //信道关闭时，要取消订阅
            basicCancel();
        }

        bool openChannel()
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个声明虚拟机的请求对象
            openChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->is_ok();
        }

        void closeChannel()
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个声明虚拟机的请求对象
            closeChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            waitResponse(rid);
            return;
        }
        
        bool declareExchange(const std::string& ename, ExchangeType type, 
            bool durable, bool auto_delete, google::protobuf::Map<std::string, std::string>& args)
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个声明虚拟机的请求对象
            declareExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->is_ok() == true)
            {
                DLOG("声明交换机成功!");
            }
            else 
            {
                DLOG("声明交换机失败!");
            }
            return resp->is_ok();
        }

        void deleteExchange(const std::string& ename)
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个删除虚拟机的请求对象
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            waitResponse(rid);
            return;
        }

        bool declareQueue(const std::string& qname, bool durable, bool exclusive, 
            bool auto_delete, google::protobuf::Map<std::string, std::string> args)
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个声明的请求对象
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_durable(durable);
            req.set_exclusive(exclusive);
            req.set_durable(auto_delete);
            req.mutable_args()->swap(args);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->is_ok() == true)
            {
                DLOG("声明队列成功!");
            }
            else 
            {
                DLOG("声明队列失败!");
            }
            return resp->is_ok();
        }

        void deleteQueue(const std::string& qname)
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个删除队列的请求对象
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            waitResponse(rid);
            return;
        }

        bool queueBind(const std::string& ename, const std::string& qname, const std::string& key)
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个队列绑定的请求对象
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(key);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->is_ok() == true)
            {
                DLOG("绑定队列成功!");
            }
            else 
            {
                DLOG("绑定队列失败!");
            }
            return resp->is_ok();
        }

        void queueUnBind(const std::string& ename, const std::string& qname)
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个队列解绑定的请求对象
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            waitResponse(rid);
            return;
        }

        void basicPublish(const std::string& ename, const BasicProperties* bp, const std::string& body)
        {
            std::string rid = UUIDHelper::uuid();
            //构造一个消息发布的请求对象
            basicPublishRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_body(body);
            if (bp)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            waitResponse(rid);
            return;
        }

        void basicAck(const std::string& msgid)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("消息确认时, 找不到消费者信息")
                return;
            }
            std::string rid = UUIDHelper::uuid();
            //构造一个消息确认的请求对象
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_message_id(msgid);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            waitResponse(rid);
            return;
        }

        bool basicConsume(const std::string& consumer_tag, const std::string& qname,
            bool auto_ack, const ConsumerCallBack& cb)
        {
            if (_consumer.get() != nullptr)
            {
                DLOG("已经订阅过队列%s, 不能重复订阅")
                return false;
            }
            std::string rid = UUIDHelper::uuid();
            //构造一个订阅队列的请求对象
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_consumer_tag(consumer_tag);
            req.set_auto_ack(auto_ack);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->is_ok() == false)
            {
                DLOG("添加订阅失败");
                return false;
            }
            _consumer = std::make_shared<Consumer>(consumer_tag, qname, auto_ack, cb);
            return true;
        }

        void basicCancel()
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("消费者不存在");
                return;
            }
            std::string rid = UUIDHelper::uuid();
            //构造一个消息确认的请求对象
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_consumer_tag(_consumer->tag);
            //向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            waitResponse(rid);
            _consumer.reset(); //重置订阅
            return;
        }
    
    public:
        //连接收到基础响应，向响应消息队列中添加数据
        void putBasicResponse(const basicCommonResponsePtr& resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert({resp->rid(), resp});
            _cv.notify_all();
        }

        //连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
        void consume(const basicConsumeResponsePtr& resp)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("消息处理时，未找到订阅者信息!");
                return;
            }

            if (_consumer->tag != resp->consumer_tag())
            {
                DLOG("收到的推送消息中消费者标识和当前信道消费者不一致!");
                return;
            }

            //_consumer->cb已经说明了收到消息后如何处理
            _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }

        std::string getCid()
        {
            return _cid;
        }
    
    private:
        basicCommonResponsePtr waitResponse(const std::string& rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [&rid, this](){
                return _basic_resp.find(rid) != _basic_resp.end();
            });

            basicCommonResponsePtr basic_resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return basic_resp;
        }

    private:
        std::string _cid;         // 1）信道 ID
        muduo::net::TcpConnectionPtr _conn;  // 2）信道关联的网络通信连接对象
        ProtobufCodecPtr _codec;     // 3）protobuf 协议处理对象
        Consumer::ptr _consumer;  // 4）信道关联的消费者
        std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;  // 5）请求对应的响应信息队列
        std::mutex _mutex;             //6）互斥锁
        std::condition_variable _cv;   //7）条件变量
    };


    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager()
        {}

        Channel::ptr create(muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            std::string cid = UUIDHelper::uuid();
            auto channel = std::make_shared<Channel>(cid, conn, codec);
            _channels.insert({cid, channel});
            return channel;
        }

        void remove(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                return;
            }
            _channels.erase(it);
        }

        Channel::ptr get(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                return {};
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}